package com.hngy.java;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;

/**
 * 需求：TopN主播统计
 * 1：首先获取两份数据中的核心字段，使用fastjson包解析数据
 * 主播开播记录(video_info.log):主播ID：uid，直播间ID：vid，大区：area
 * (vid,(uid,area))
 * 用户送礼记录(gift_record.log)：直播间ID：vid，金币数量：gold
 * (vid,gold)
 *
 * 这样的话可以把这两份数据关联到一块就能获取到大区、主播id、金币这些信息了，使用直播
 *
 * 2：对用户送礼记录数据进行聚合，对相同vid的数据求和
 * (vid,gold_sum)
 *
 * 3：把这两份数据join到一块，vid作为join的key
 * (vid,((uid,area),gold_sum))
 *
 * 4：使用map迭代join之后的数据，最后获取到uid、area、gold_sum字段
 * 由于一个主播一天可能会开播多次，后面需要基于uid和area再做一次聚合，所以把数据转换
 *
 * uid和area是一一对应的，一个人只能属于大区
 * ((uid,area),gold_sum)
 *
 * 5：使用reduceByKey算子对数据进行聚合
 * ((uid,area),gold_sum_all)
 *
 * 6：接下来对需要使用groupByKey对数据进行分组，所以先使用map进行转换
 * 因为我们要分区统计TopN，所以要根据大区分组
 * map：(area,(uid,gold_sum_all))
 * groupByKey: area,<(uid,gold_sum_all),(uid,gold_sum_all),(uid,gold_sum_all)
 *
 * 7：使用map迭代每个分组内的数据，按照金币数量倒序排序，取前N个，最终输出area,topN
 * 这个topN其实就是把前几名主播的id还有金币数量拼接成一个字符串
 * (area,topN)
 *
 * 8：使用foreach将结果打印到控制台，多个字段使用制表符分割
 * area topN
 */
public class TopNSJava {

    public static void main(String[] args){
        SparkConf conf = new SparkConf();
        conf.setAppName("TopNSJava").setMaster("local");

        //创建SparkSession对象，里面包含SparkContext和SqlContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        //1：首先获取两份数据中的核心字段，使用fastjson包解析数据
        JavaRDD<String> videoInfoRDD = sc.textFile("F:\\BaiduNetdiskDownload\\hadoop\\source\\bigdata_course_materials\\spark1\\video_info.log");
        JavaRDD<String> giftRecordRDD = sc.textFile("F:\\BaiduNetdiskDownload\\hadoop\\source\\bigdata_course_materials\\spark1\\gift_record.log");

        //(vid,(uid,area))
        JavaPairRDD<String, Tuple2<String, String>> videoInfoFieldRDD = videoInfoRDD.mapToPair(new PairFunction<String, String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, Tuple2<String, String>> call(String line) {
                JSONObject jsonObject = JSON.parseObject(line);
                String vid = jsonObject.getString("vid");
                String uid = jsonObject.getString("uid");
                String area = jsonObject.getString("area");
                return new Tuple2<>(vid, new Tuple2<>(uid, area));
            }
        });

        //(vid,gold)
        JavaPairRDD<String, Integer> giftRecordFieldRDD = giftRecordRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String line) {
                JSONObject jsonObject = JSON.parseObject(line);
                String vid = jsonObject.getString("vid");
                int gold = Integer.parseInt(jsonObject.getString("gold"));
                return new Tuple2<>(vid, gold);
            }
        });

        //2：对用户送礼记录数据进行聚合，对相同vid的数据求和
        //(vid,gold_sum)
        JavaPairRDD<String, Integer> giftRecordFieldAggRDD = giftRecordFieldRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer i1, Integer i2) throws Exception {
                return i1 + i2;
            }
        });

        //3：把这两份数据join到一块，vid作为join的key
        //(vid,((uid,area),gold_sum))
        JavaPairRDD<String, Tuple2<Tuple2<String, String>, Integer>> joinRDD = videoInfoFieldRDD.join(giftRecordFieldAggRDD);

        //4：使用map迭代join之后的数据，最后获取到uid、area、gold_sum字段
        //((uid,area),gold_sum)
        JavaPairRDD<Tuple2<String, String>, Integer> joinMapRDD = joinRDD.mapToPair(new PairFunction<Tuple2<String, Tuple2<Tuple2<String, String>, Integer>>, Tuple2<String, String>, Integer>() {
            @Override
            public Tuple2<Tuple2<String, String>, Integer> call(Tuple2<String, Tuple2<Tuple2<String, String>, Integer>> tup) {
                String area = tup._2._1._2;
                String uid = tup._2._1._1;
                int gold_sum = tup._2._2;
                return new Tuple2<>(new Tuple2<>(uid, area), gold_sum);
            }
        });

        //5：使用reduceByKey算子对数据进行聚合
        //((uid,area),gold_sum_all)
        JavaPairRDD<Tuple2<String, String>, Integer> joinMapAggRDD = joinMapRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer i1, Integer i2) throws Exception {
                return i1 + i2;
            }
        });

        //6：接下来对需要使用groupByKey对数据进行分组，所以先使用map进行转换
        //map：(area,(uid,gold_sum_all))
        //groupByKey: area,<(uid,gold_sum_all),(uid,gold_sum_all),(uid,gold_sum_a
        JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> groupRDD = joinMapAggRDD.mapToPair(new PairFunction<Tuple2<Tuple2<String, String>, Integer>, String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Tuple2<String, Integer>> call(Tuple2<Tuple2<String, String>, Integer> tup) {
                String area = tup._1._2;
                String uid = tup._1._1;
                int gold_sum_all = tup._2;
                return new Tuple2<>(area, new Tuple2<>(uid, gold_sum_all));
            }
        }).groupByKey();


        //7：使用map迭代每个分组内的数据，按照金币数量倒序排序，取前N个，最终输出area,t
        //(area,topN)
        JavaRDD<Tuple2<String, String>> top3RDD = groupRDD.map(new Function<Tuple2<String, Iterable<Tuple2<String, Integer>>>, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> call(Tuple2<String, Iterable<Tuple2<String, Integer>>> tup) {
                String area = tup._1;
                ArrayList<Tuple2<String, Integer>> tupList = Lists.newArrayList(tup._2);
                //对集合中的元素排序
                Collections.sort(tupList, new Comparator<Tuple2<String, Integer>>() {
                    @Override
                    public int compare(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) {
                        return t2._2 - t1._2;
                    }
                });
                //uid:gold_sum_all,uid:gold_sum_all,uid:gold_sum_all
                StringBuffer sb = new StringBuffer();
                for (int i = 0; i < tupList.size(); i++) {
                    if (i < 3) { //top 3
                        Tuple2<String, Integer> t = tupList.get(i);
                        if (i != 0) {
                            sb.append(",");
                        }
                        sb.append(t._1 + ":" + t._2);
                    }
                }
                return new Tuple2<>(area, sb.toString());
            }
        });

        //8：使用foreach将结果打印到控制台，多个字段使用制表符分割
        top3RDD.foreach(new VoidFunction<Tuple2<String, String>>() {
            @Override
            public void call(Tuple2<String, String> tup) throws Exception {
                System.out.println(tup._1 + "\t" + tup._2);
            }
        });

        sc.stop();
    }
}
